"""
Case Type   : 基础功能
Case Name   : 同步基础数据时,表存在唯一约束
Create At   : 2022/11/09
Owner       : opentestcase026
Description :
    1.两个集群均创建表1与表2
    2.发布端在表1和表2内插入数据
    3.订阅端在表2内插入数据
    4.创建发布端
    5.创建订阅端
    6.查询订阅端表1与表2数据
    7.在订阅端修改表1内数据
    8.查询订阅端表1与表2数据
    9.删除订阅
    10.删除发布
    11.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.表1内数据未同步,表2内数据同步
    7.成功
    8.表1与表2内数据均同步
    9.成功
    10.成功
    11.成功
History     :
"""
import unittest
import os

from yat.test import macro
from yat.test import Node

from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != primary_sh.get_node_num(), '非1+2环境不执行')
class PubSubClass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----this is setup-----")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_user_pub = Node(node='PrimaryDbUser')
        self.pri_user_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.com_sh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name1 = 'tb_pub_sub_case249_1'
        self.tb_name2 = 'tb_pub_sub_case249_2'
        self.sub_name = "sub_case249"
        self.pub_name = "pub_case249"
        self.port = int(self.pri_user_pub.db_port) + 1
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_user_pub.db_user} ' \
            f'-W {self.pri_user_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_user_sub.db_user} ' \
            f'-W {self.pri_user_sub.db_password}'
        text = '----修改pg_hba expect:成功----'
        self.log.info(text)
        self.log.info('----发布端状态----')
        status = primary_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        self.log.info('----订阅端状态----')
        status = \
            self.com_sh_sub.get_db_cluster_status(
                'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        guc_res = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = primary_sh.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '-----step1:两个集群均创建表1与表2（发布端与订阅端两表结构不一致） ' \
               'expect:成功-----'
        self.log.info(text)
        sql = f"drop table if exists {self.tb_name1};" \
              f"create table {self.tb_name1}" \
              f"(i int primary key, t text UNIQUE);" \
              f"drop table if exists {self.tb_name2};" \
              f"create table {self.tb_name2}" \
              f"(i int primary key, t text UNIQUE," \
              f"FOREIGN KEY(i) REFERENCES {self.tb_name1}(i));"
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         6, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         6, '执行失败:' + text)

        text = '-----step2:发布端在表1与表2内插入数据 expect:成功-----'
        self.log.info(text)
        sql1 = f"insert into {self.tb_name1} values(1,'pub_sub');" \
               f"select pg_sleep(5.5);" \
               f"insert into {self.tb_name2} values(1,'pub');" \
               f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql1,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '-----step3:订阅端在表1内插入数据 expect:成功-----'
        self.log.info(text)
        sql2 = f"insert into {self.tb_name1} values(2,'pub_sub');" \
               f"select pg_sleep(5.5);"
        result = self.com_sh_sub.execut_db_sql(sql2, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '-----step4:创建发布端;expect:成功-----'
        self.log.info(text)
        sql = f"drop publication if exists " \
              f"{self.pub_name};" \
              f"create publication {self.pub_name} " \
              f"for all tables;" \
              f"select pg_sleep(3);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.drop_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)

        text = '-----step5:创建订阅端;expect:成功-----'
        self.log.info(text)
        sql = f"drop subscription if exists {self.sub_name};" \
              f"create subscription {self.sub_name} connection " \
              f"'host={self.pri_user_pub.db_host} " \
              f"port={self.port} " \
              f"user={self.pri_user_pub.db_user} " \
              f"dbname={self.pri_user_pub.db_name} " \
              f"password={self.pri_user_pub.ssh_password}' " \
              f"publication {self.pub_name} with (copy_data=true);" \
              f"select pg_sleep(5.5);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.drop_sub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        text = '-----step6:查询表1与表2数据; ' \
               'expect:表1数据未同步（表1内只有在订阅端口插入的数据）,表2数据同步-----'
        self.log.info(text)
        sql1 = f"select * from {self.tb_name1};"
        result1 = self.com_sh_sub.execut_db_sql(sql1, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result1)
        self.assertEqual(result1.count('1 row'), 1, '执行失败' + text)
        self.assertIn('2 | pub_sub', result1, '执行失败:' + text)
        sql2 = f"select * from {self.tb_name2};"
        result2 = self.com_sh_sub.execut_db_sql(sql2, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertEqual(result2.count('1 row'), 1, '执行失败' + text)
        self.assertIn('1 | pub', result2, '执行失败:' + text)

        text = '-----step7:在订阅端修改表1内数据 expect:成功-----'
        self.log.info(text)
        sql = f"update {self.tb_name1} set t='update' where i=2;" \
              f"select pg_sleep(5.5);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, result,
                      '执行失败:' + text)

        text = '-----step8:查询表1与表2内数据 expect:数据同步-----'
        self.log.info(text)
        sql1 = f"select * from {self.tb_name1};"
        result1 = self.com_sh_sub.execut_db_sql(sql1, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result1)
        self.assertEqual(result1.count('2 rows'), 1, '执行失败' + text)
        self.assertIn('1 | pub_sub', result1, '执行失败:' + text)
        sql2 = f"select * from {self.tb_name2};"
        result2 = self.com_sh_sub.execut_db_sql(sql2, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertEqual(result2.count('1 row'), 1, '执行失败' + text)
        self.assertIn('1 | pub', result2, '执行失败:' + text)

    def tearDown(self):
        self.log.info('-----this is tearDown-----')
        text = '-----step9:删除订阅 expect:成功-----'
        self.log.info(text)
        sql = f"drop subscription  {self.sub_name};"
        drop_sub_result = self.com_sh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)

        text = '-----step10:删除发布 expect:成功-----'
        self.log.info(text)
        sql = f"drop publication {self.pub_name};"
        drop_pub_result = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text = '-----step11:清理环境,删除表 expect:成功-----'
        self.log.info(text)
        sql = f"drop table {self.tb_name1},{self.tb_name2};"
        result1 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result1)
        result2 = primary_sh.execut_db_sql(sql,
                                           sql_type=self.user_param_pub)
        self.log.info(result2)
        guc_res2 = primary_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(guc_res2)
        guc_res1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32')
        self.log.info(guc_res1)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        result = primary_sh.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result \
              or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertTrue(guc_res, '执行失败:' + text)
        self.assertTrue(guc_res1, '执行失败:' + text)
        self.assertTrue(guc_res2, '执行失败:' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")
